package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"github.com/louloulin/dataflare/pkg/processing/data"
	"github.com/louloulin/dataflare/pkg/processing/election"
	"github.com/louloulin/dataflare/pkg/processing/replication"
)

// Node 表示节点
type Node struct {
	// ID 是节点ID
	ID string
	// replicator 是复制器
	replicator *replication.Replicator
	// election 是选举
	election *election.LeaderElection
	// data 是数据存储
	data map[string][]byte
	// mu 是互斥锁
	mu sync.RWMutex
}

// NewNode 创建一个新的节点
func NewNode(id string, role replication.ReplicationRole, partners []string) *Node {
	// 创建复制传输
	replicationTransport := &MockReplicationTransport{
		nodeID:   id,
		nodes:    make(map[string]*Node),
		batches:  make(map[string][]*data.DataBatch),
		acks:     make(map[string]uint64),
		batchCh:  make(chan *BatchMessage, 100),
		ackCh:    make(chan *AckMessage, 100),
	}

	// 创建复制配置
	replicationConfig := &replication.ReplicationConfig{
		Mode:          replication.ReplicationModeAsync,
		Role:          role,
		GroupID:       "test-group",
		NodeID:        id,
		Partners:      partners,
		SyncInterval:  time.Second,
		MaxBatchSize:  10,
		AckTimeout:    time.Second * 5,
		RetryCount:    3,
		RetryInterval: time.Second,
	}

	// 创建复制器
	replicator := replication.NewReplicator(replicationConfig, replicationTransport)

	// 创建选举传输
	electionTransport := &MockElectionTransport{
		nodeID:        id,
		nodes:         make(map[string]*Node),
		voteRequests:  make(chan *VoteRequest, 100),
		voteResponses: make(chan *VoteResponse, 100),
		heartbeats:    make(chan *Heartbeat, 100),
	}

	// 创建选举配置
	electionConfig := &election.ElectionConfig{
		NodeID:           id,
		ClusterID:        "test-cluster",
		ElectionTimeout:  time.Second * 5,
		HeartbeatInterval: time.Second,
		MinVotes:         len(partners) / 2 + 1,
	}

	// 创建选举
	electionObj := election.NewLeaderElection(electionConfig, electionTransport)

	// 创建节点
	node := &Node{
		ID:         id,
		replicator: replicator,
		election:   electionObj,
		data:       make(map[string][]byte),
	}

	// 设置传输节点引用
	replicationTransport.node = node
	electionTransport.node = node

	return node
}

// Start 启动节点
func (n *Node) Start() error {
	// 启动复制器
	err := n.replicator.Start()
	if err != nil {
		return fmt.Errorf("failed to start replicator: %w", err)
	}

	// 启动选举
	err = n.election.Start()
	if err != nil {
		n.replicator.Stop()
		return fmt.Errorf("failed to start election: %w", err)
	}

	return nil
}

// Stop 停止节点
func (n *Node) Stop() error {
	// 停止选举
	err := n.election.Stop()
	if err != nil {
		return fmt.Errorf("failed to stop election: %w", err)
	}

	// 停止复制器
	err = n.replicator.Stop()
	if err != nil {
		return fmt.Errorf("failed to stop replicator: %w", err)
	}

	return nil
}

// Put 存储数据
func (n *Node) Put(key string, value []byte) error {
	// 检查是否是领导者
	if !n.election.IsLeader() {
		return fmt.Errorf("not leader")
	}

	// 存储数据
	n.mu.Lock()
	n.data[key] = value
	n.mu.Unlock()

	// 创建记录
	record := data.NewDataRecord([]byte(key), value, data.DataFormatBinary)

	// 复制记录
	err := n.replicator.Replicate(record)
	if err != nil {
		return fmt.Errorf("failed to replicate record: %w", err)
	}

	return nil
}

// Get 获取数据
func (n *Node) Get(key string) ([]byte, bool) {
	n.mu.RLock()
	defer n.mu.RUnlock()
	value, ok := n.data[key]
	return value, ok
}

// Delete 删除数据
func (n *Node) Delete(key string) error {
	// 检查是否是领导者
	if !n.election.IsLeader() {
		return fmt.Errorf("not leader")
	}

	// 删除数据
	n.mu.Lock()
	delete(n.data, key)
	n.mu.Unlock()

	// 创建记录
	record := data.NewDataRecord([]byte(key), nil, data.DataFormatBinary)
	record.SetHeader("deleted", "true")

	// 复制记录
	err := n.replicator.Replicate(record)
	if err != nil {
		return fmt.Errorf("failed to replicate record: %w", err)
	}

	return nil
}

// IsLeader 检查是否是领导者
func (n *Node) IsLeader() bool {
	return n.election.IsLeader()
}

// GetLeader 获取领导者
func (n *Node) GetLeader() string {
	return n.election.GetCurrentLeader()
}

// GetState 获取状态
func (n *Node) GetState() string {
	return n.election.GetState().String()
}

// GetTerm 获取任期
func (n *Node) GetTerm() uint64 {
	return n.election.GetCurrentTerm()
}

// BatchMessage 表示批次消息
type BatchMessage struct {
	// SenderID 是发送者ID
	SenderID string
	// ReceiverID 是接收者ID
	ReceiverID string
	// Batch 是数据批次
	Batch *data.DataBatch
}

// AckMessage 表示确认消息
type AckMessage struct {
	// SenderID 是发送者ID
	SenderID string
	// ReceiverID 是接收者ID
	ReceiverID string
	// Offset 是偏移量
	Offset uint64
}

// MockReplicationTransport 模拟复制传输
type MockReplicationTransport struct {
	// nodeID 是节点ID
	nodeID string
	// node 是节点引用
	node *Node
	// nodes 是节点映射表
	nodes map[string]*Node
	// batches 是批次映射表
	batches map[string][]*data.DataBatch
	// acks 是确认映射表
	acks map[string]uint64
	// batchCh 是批次通道
	batchCh chan *BatchMessage
	// ackCh 是确认通道
	ackCh chan *AckMessage
	// mu 是互斥锁
	mu sync.RWMutex
}

// Connect 连接到伙伴
func (t *MockReplicationTransport) Connect(ctx context.Context, partnerID string, address string) error {
	return nil
}

// Disconnect 断开与伙伴的连接
func (t *MockReplicationTransport) Disconnect(ctx context.Context, partnerID string) error {
	return nil
}

// SendBatch 发送批次
func (t *MockReplicationTransport) SendBatch(ctx context.Context, partnerID string, batch *data.DataBatch) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	// 发送批次消息
	select {
	case t.batchCh <- &BatchMessage{
		SenderID:   t.nodeID,
		ReceiverID: partnerID,
		Batch:      batch,
	}:
	case <-ctx.Done():
		return ctx.Err()
	}

	return nil
}

// ReceiveBatch 接收批次
func (t *MockReplicationTransport) ReceiveBatch(ctx context.Context) (*data.DataBatch, string, error) {
	select {
	case msg := <-t.batchCh:
		if msg.ReceiverID == t.nodeID {
			return msg.Batch, msg.SenderID, nil
		}
		// 转发消息
		t.batchCh <- msg
		// 等待下一个消息
		return t.ReceiveBatch(ctx)
	case <-ctx.Done():
		return nil, "", ctx.Err()
	}
}

// SendAck 发送确认
func (t *MockReplicationTransport) SendAck(ctx context.Context, partnerID string, offset uint64) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	// 发送确认消息
	select {
	case t.ackCh <- &AckMessage{
		SenderID:   t.nodeID,
		ReceiverID: partnerID,
		Offset:     offset,
	}:
	case <-ctx.Done():
		return ctx.Err()
	}

	return nil
}

// ReceiveAck 接收确认
func (t *MockReplicationTransport) ReceiveAck(ctx context.Context) (string, uint64, error) {
	select {
	case msg := <-t.ackCh:
		if msg.ReceiverID == t.nodeID {
			return msg.SenderID, msg.Offset, nil
		}
		// 转发消息
		t.ackCh <- msg
		// 等待下一个消息
		return t.ReceiveAck(ctx)
	case <-ctx.Done():
		return "", 0, ctx.Err()
	}
}

// VoteRequest 表示投票请求
type VoteRequest struct {
	// Term 是任期
	Term uint64
	// CandidateID 是候选者ID
	CandidateID string
	// ResponseCh 是响应通道
	ResponseCh chan *VoteResponse
}

// VoteResponse 表示投票响应
type VoteResponse struct {
	// Granted 是是否授予投票
	Granted bool
	// Term 是任期
	Term uint64
}

// Heartbeat 表示心跳
type Heartbeat struct {
	// Term 是任期
	Term uint64
	// LeaderID 是领导者ID
	LeaderID string
}

// MockElectionTransport 模拟选举传输
type MockElectionTransport struct {
	// nodeID 是节点ID
	nodeID string
	// node 是节点引用
	node *Node
	// nodes 是节点映射表
	nodes map[string]*Node
	// voteRequests 是投票请求通道
	voteRequests chan *VoteRequest
	// voteResponses 是投票响应通道
	voteResponses chan *VoteResponse
	// heartbeats 是心跳通道
	heartbeats chan *Heartbeat
	// mu 是互斥锁
	mu sync.RWMutex
}

// RequestVote 请求投票
func (t *MockElectionTransport) RequestVote(ctx context.Context, term uint64, candidateID string) (bool, uint64, error) {
	// 创建响应通道
	responseCh := make(chan *VoteResponse, 1)

	// 发送请求
	select {
	case t.voteRequests <- &VoteRequest{
		Term:        term,
		CandidateID: candidateID,
		ResponseCh:  responseCh,
	}:
	case <-ctx.Done():
		return false, 0, ctx.Err()
	}

	// 等待响应
	select {
	case resp := <-responseCh:
		return resp.Granted, resp.Term, nil
	case <-ctx.Done():
		return false, 0, ctx.Err()
	}
}

// SendHeartbeat 发送心跳
func (t *MockElectionTransport) SendHeartbeat(ctx context.Context, term uint64, leaderID string) error {
	// 发送心跳
	select {
	case t.heartbeats <- &Heartbeat{
		Term:     term,
		LeaderID: leaderID,
	}:
	case <-ctx.Done():
		return ctx.Err()
	}

	return nil
}

// BroadcastLeader 广播领导者
func (t *MockElectionTransport) BroadcastLeader(ctx context.Context, term uint64, leaderID string) error {
	// 发送心跳
	select {
	case t.heartbeats <- &Heartbeat{
		Term:     term,
		LeaderID: leaderID,
	}:
	case <-ctx.Done():
		return ctx.Err()
	}

	return nil
}

// ElectionEventHandler 选举事件处理器
type ElectionEventHandler struct {
	// node 是节点引用
	node *Node
}

// HandleElectionEvent 处理选举事件
func (h *ElectionEventHandler) HandleElectionEvent(event election.ElectionEvent, data map[string]interface{}) {
	switch event {
	case election.ElectionEventBecomeLeader:
		log.Printf("Node %s became leader for term %d", h.node.ID, data["term"])
	case election.ElectionEventBecomeFollower:
		log.Printf("Node %s became follower for term %d", h.node.ID, data["term"])
	case election.ElectionEventBecomeCandidate:
		log.Printf("Node %s became candidate for term %d", h.node.ID, data["term"])
	case election.ElectionEventLeaderChanged:
		log.Printf("Node %s detected leader change to %s for term %d", h.node.ID, data["leader_id"], data["term"])
	case election.ElectionEventTermChanged:
		log.Printf("Node %s detected term change from %d to %d", h.node.ID, data["old_term"], data["new_term"])
	}
}

func main() {
	// 创建节点
	node1 := NewNode("node-1", replication.ReplicationRolePrimary, []string{"node-2", "node-3"})
	node2 := NewNode("node-2", replication.ReplicationRoleReplica, []string{"node-1", "node-3"})
	node3 := NewNode("node-3", replication.ReplicationRoleReplica, []string{"node-1", "node-2"})

	// 设置节点引用
	node1.replicator.(*replication.Replicator).Transport().(*MockReplicationTransport).nodes["node-2"] = node2
	node1.replicator.(*replication.Replicator).Transport().(*MockReplicationTransport).nodes["node-3"] = node3
	node2.replicator.(*replication.Replicator).Transport().(*MockReplicationTransport).nodes["node-1"] = node1
	node2.replicator.(*replication.Replicator).Transport().(*MockReplicationTransport).nodes["node-3"] = node3
	node3.replicator.(*replication.Replicator).Transport().(*MockReplicationTransport).nodes["node-1"] = node1
	node3.replicator.(*replication.Replicator).Transport().(*MockReplicationTransport).nodes["node-2"] = node2

	node1.election.(*election.LeaderElection).Transport().(*MockElectionTransport).nodes["node-2"] = node2
	node1.election.(*election.LeaderElection).Transport().(*MockElectionTransport).nodes["node-3"] = node3
	node2.election.(*election.LeaderElection).Transport().(*MockElectionTransport).nodes["node-1"] = node1
	node2.election.(*election.LeaderElection).Transport().(*MockElectionTransport).nodes["node-3"] = node3
	node3.election.(*election.LeaderElection).Transport().(*MockElectionTransport).nodes["node-1"] = node1
	node3.election.(*election.LeaderElection).Transport().(*MockElectionTransport).nodes["node-2"] = node2

	// 添加事件处理器
	node1.election.AddEventHandler(&ElectionEventHandler{node: node1})
	node2.election.AddEventHandler(&ElectionEventHandler{node: node2})
	node3.election.AddEventHandler(&ElectionEventHandler{node: node3})

	// 启动节点
	log.Println("Starting nodes...")
	err := node1.Start()
	if err != nil {
		log.Fatalf("Failed to start node1: %v", err)
	}
	err = node2.Start()
	if err != nil {
		log.Fatalf("Failed to start node2: %v", err)
	}
	err = node3.Start()
	if err != nil {
		log.Fatalf("Failed to start node3: %v", err)
	}

	// 等待选举完成
	log.Println("Waiting for election to complete...")
	time.Sleep(time.Second * 2)

	// 打印节点状态
	log.Printf("Node 1: state=%s, term=%d, leader=%s", node1.GetState(), node1.GetTerm(), node1.GetLeader())
	log.Printf("Node 2: state=%s, term=%d, leader=%s", node2.GetState(), node2.GetTerm(), node2.GetLeader())
	log.Printf("Node 3: state=%s, term=%d, leader=%s", node3.GetState(), node3.GetTerm(), node3.GetLeader())

	// 找到领导者
	var leader *Node
	if node1.IsLeader() {
		leader = node1
	} else if node2.IsLeader() {
		leader = node2
	} else if node3.IsLeader() {
		leader = node3
	} else {
		log.Fatal("No leader found")
	}

	// 写入数据
	log.Printf("Writing data to leader %s...", leader.ID)
	err = leader.Put("key1", []byte("value1"))
	if err != nil {
		log.Fatalf("Failed to put data: %v", err)
	}
	err = leader.Put("key2", []byte("value2"))
	if err != nil {
		log.Fatalf("Failed to put data: %v", err)
	}

	// 等待复制完成
	log.Println("Waiting for replication to complete...")
	time.Sleep(time.Second * 2)

	// 读取数据
	log.Println("Reading data from all nodes...")
	for _, node := range []*Node{node1, node2, node3} {
		value1, ok1 := node.Get("key1")
		value2, ok2 := node.Get("key2")
		log.Printf("Node %s: key1=%v (%v), key2=%v (%v)", node.ID, string(value1), ok1, string(value2), ok2)
	}

	// 删除数据
	log.Printf("Deleting data from leader %s...", leader.ID)
	err = leader.Delete("key1")
	if err != nil {
		log.Fatalf("Failed to delete data: %v", err)
	}

	// 等待复制完成
	log.Println("Waiting for replication to complete...")
	time.Sleep(time.Second * 2)

	// 读取数据
	log.Println("Reading data from all nodes...")
	for _, node := range []*Node{node1, node2, node3} {
		value1, ok1 := node.Get("key1")
		value2, ok2 := node.Get("key2")
		log.Printf("Node %s: key1=%v (%v), key2=%v (%v)", node.ID, string(value1), ok1, string(value2), ok2)
	}

	// 等待信号
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	log.Println("Press Ctrl+C to stop...")
	<-sigCh

	// 停止节点
	log.Println("Stopping nodes...")
	node1.Stop()
	node2.Stop()
	node3.Stop()
}
